package tsif.tcluster.service.proxy.impl;

import java.lang.reflect.Constructor;
import java.util.Arrays;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import tsif.tcluster.ITClusterAccessor;
import tsif.tcluster.TClusterContext;
import tsif.tcluster.rpc.TClusterRpcCallBackMsg;
import tsif.tcluster.rpc.TClusterRpcCallMsg;
import tsif.tcluster.rpc.TClusterRpcProtocol;
import tsif.tcluster.rpc.TClusterRpcUtils;
import tsif.tcluster.service.ITClusterService;
import tsif.tcluster.service.ITClusterService.IReturn;
import tsif.tcluster.service.proxy.TClusterServiceProxy;
import tsif.tcluster.service.proxy.impl.TClusterCallbacks.CallBack;
import tsif.tcluster.session.TClusterSession;

/**
 * 集群云服务代理<br>
 * 控制单个服务的业务更新
 * 
 */
public class TClusterRpcServiceProxy extends TClusterServiceProxy {
	protected TClusterContext context; // 环境

	protected TClusterCallbacks callbacks;
	protected TClusterAccessor accessor;

	protected ExecutorService executor; // 服务的线程池

	public TClusterRpcServiceProxy() {
		callbacks = new TClusterCallbacks(this);
		accessor = new TClusterAccessor(this);
	}

	@Override
	protected boolean init(TClusterContext context, long serviceId, String uniqueName, Class<?> classValue, Object... args) throws Exception {
		this.serviceId = serviceId;
		this.uniqueName = uniqueName;
		this.context = context;
		// 加载服务
		service = createService(uniqueName, classValue, args);
		if (service == null) {
			throw new Exception("创建服务失败! " + uniqueName + " " + classValue);
		}
		executor = new TClusterExecutorService(classValue.getSimpleName());
		// this.executor = Executors.newCachedThreadPool();
		// 初始化服务
		if (!service.init(this, serviceId, args)) {
			throw new Exception("初始化服务失败! " + uniqueName + " " + classValue);
		}
		return true;
	}

	@Override
	protected void destroy() throws Exception {
		if (service != null) {
			service.destroy(); // 销毁处理
			service = null;
		}
		if (executor != null) {
			executor.shutdown();
			executor.awaitTermination(1, TimeUnit.MINUTES);
			executor = null;
		}
	}

	/** 获取类名 **/
	public static String getClassName(Object classValue) {
		if (classValue instanceof String) {
			// 字符串/类名, 用类名加载类
			return (String) classValue;
		} else if (classValue instanceof Class) {
			// 类对象, 换成类名再加载
			return ((Class<?>) classValue).getName();
		}
		return null;
	}

	/** 类加载器 **/
	protected ClassLoader getClassLoader() {
		return this.getClass().getClassLoader();
	}

	/** 创建服务 **/
	protected ITClusterService createService(String name, Object classValue, Object... args) throws Exception {
		ClassLoader classLoader = getClassLoader();

		// get class name
		String className = getClassName(classValue);
		if (className == null || className.length() <= 0) {
			throw new Exception("unkown class: " + classValue);
		}
		Class<?> clazz = classLoader.loadClass(className);

		// create
		Constructor<?> constructor = clazz.getDeclaredConstructor(new Class<?>[] {});
		constructor.setAccessible(true);
		ITClusterService service = (ITClusterService) constructor.newInstance();
		return service;
	}

	@Override
	protected void update() {
		// 更新回调
		this.callbacks.update();

		// 更新服务
		if (service != null) {
			this.service.update();
		}
	}

	/** 获取执行线程池(默认获取环境的线程池) **/
	public Executor getExecutor() {
		return this.executor;
	}

	/** rpc调用 **/
	protected void dispose(final long serviceId, final String funcName, final Object[] args, final IReturn<Object> ret) throws Exception {
		// 提交到线程池
		execute(new Runnable() {
			@Override
			public void run() {
				try {
					service.dispose(serviceId, funcName, args, ret);
				} catch (RuntimeException e) {
					throw e;
				} catch (Exception e) {
					throw new RuntimeException(e);
				}
			}
		});
	}

	/** 提交任务 **/
	public void execute(Runnable command) {
		executor.execute(command); // 提交线程
		// command.run(); // 直接执行
	}

	/** 触发调用 **/
	protected void onCall(final TClusterSession session, final long serviceId, TClusterRpcCallMsg msg) throws Exception {
		final int callbackId = msg.getCallbackId();
		try {
			final String funcName = msg.getMethod();
			final Object[] args = decodeAutoData(msg.getArgDatas()); // 实际上现在这个是没有用处的(在上一步解析的过程就已经完成了)

			// create callback obj
			IReturn<Object> ret = new IReturn<Object>() {
				protected boolean close = false;

				@Override
				public void ret(int code, String msg, Object retObj) {
					if (close) {
						throw new RuntimeException("callback is close! " + funcName + " " + Arrays.toString(args));
					}
					close = true;

					// callback
					if (callbackId <= 0) {
						return; // no callback id
					}
					// System.out.println("ret: " + TClusterServiceProxy.this.serviceId + " -> " + serviceId + " " + funcName + "(" + callbackId + ")" + code + " " + msg + " " + retObj);
					accessor.callback(session, serviceId, callbackId, code, msg, retObj);
				}
			};

			// dispose
			// System.out.println("dispose: " + serviceId + " -> " + this.serviceId + " " + funcName + " " + Arrays.toString(args));
			dispose(serviceId, funcName, args, ret);
		} catch (Throwable e) {
			// error callback
			if (callbackId > 0) {
				accessor.callback(session, serviceId, callbackId, -1, "error: " + e.toString(), null);
			}
			// error
			throw e;
		}
	}

	/** 触发回调 **/
	protected void onCallback(TClusterSession session, long serviceId, TClusterRpcCallBackMsg msg) throws Exception {
		// System.out.println("callback: " + serviceId + " -> " + this.serviceId + " " + msg.getCallbackId());
		// get ret obj
		Object retObj = decodeAutoData(msg.getRetDatas());
		// callback
		callbacks.callback(session, msg.getCallbackId(), msg.getCode(), msg.getMsg(), retObj);
	}

	/** 解析自动类型数据 **/
	@SuppressWarnings("unchecked")
	protected <T> T decodeAutoData(Object data) throws Exception {
		if (data == null) {
			return null; // 允许空对象
		}
		// 检测是否是这个类(如果数据本身就是byte[]那就尴尬了, 再者这个没有意义, 统一在rpc消息那里就解析了)
		// if (data.getClass() == byte[].class) {
		// return (T) TClusterRpcUtils.toAutoObject((byte[]) data);
		// }
		return (T) data;
	}

	/** 解析数据 **/
	@SuppressWarnings("unchecked")
	protected <T> T decodeData(Object data, Class<?> clazz) throws Exception {
		if (data == null) {
			throw new Exception("decode error! empty data!"); // 这里是指定类型的, 不能为空
		}
		// 检测是否是这个类
		if (clazz.isInstance(data)) {
			return (T) data;
		} else if (data.getClass() == byte[].class) {
			return (T) TClusterRpcUtils.toObject((byte[]) data, clazz);
		}
		throw new Exception("decode error! class=" + data.getClass() + "/" + data.getClass().getClassLoader() + " To " + clazz + "/" + clazz.getClassLoader());
	}

	/** 添加返回函数, 返回回调Id **/
	protected <T> int addCallBack(CallBack<T> callback) {
		return this.callbacks.addCallBack(callback);
	}

	@Override
	public void recv(TClusterSession session, long serviceId, short code, Object data) throws Exception {
		try {
			// rpc read
			if (onRpcRecv(session, serviceId, code, data)) {
				return;
			}
			// other code
			if (code == TClusterRpcProtocol.RPC_CALL) {
				TClusterRpcCallMsg msg = decodeData(data, TClusterRpcCallMsg.class);
				onCall(session, serviceId, msg);
			} else if (code == TClusterRpcProtocol.RPC_CALLBACK) {
				TClusterRpcCallBackMsg msg = decodeData(data, TClusterRpcCallBackMsg.class);
				onCallback(session, serviceId, msg);
			} else {
				System.err.println("未知rpc消息类型: code=" + code);
			}
		} catch (Throwable e) {
			onRecvError(session, serviceId, code, data, e);
		}
	}

	/** rpc消息接收额外处理 **/
	protected boolean onRpcRecv(TClusterSession session, long serviceId, short code, Object data) {
		return false;
	}

	/** 消息处理错误处理 **/
	protected void onRecvError(TClusterSession session, long serviceId, short code, Object data, Throwable e) {
		System.err.println("msg action error! code=" + code);
		e.printStackTrace();
	}

	/** 获取程序 **/
	public TClusterContext getContext() {
		return context;
	}

	@Override
	public ITClusterAccessor getAccessor() {
		return accessor;
	}

	/** 服务线程(单线程) **/
	protected static class TClusterExecutorService extends ThreadPoolExecutor {

		public TClusterExecutorService(String name) {
			super(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new DefaultThreadFactory(name));
		}

		/** 执行函数(由于是单线程所以要检测运行过久的任务) **/
		// @Override
		// public void execute(final Runnable command) {
		// super.execute(new Runnable() {
		// @Override
		// public void run() {
		// long startTime = System.nanoTime();
		// }
		//
		// });
		// }
	}

	/** 默认线程工厂, 用于创建线程生成名字 **/
	protected static class DefaultThreadFactory implements ThreadFactory {
		private final ThreadGroup group;
		private final AtomicInteger threadNumber = new AtomicInteger(1);
		private final String namePrefix;

		public DefaultThreadFactory(String poolName) {
			SecurityManager securityManager = System.getSecurityManager();
			group = (securityManager != null) ? securityManager.getThreadGroup() : Thread.currentThread().getThreadGroup();
			namePrefix = "" + poolName + "-";
		}

		@Override
		public Thread newThread(Runnable run) {
			Thread thread = new Thread(group, run, namePrefix + threadNumber.getAndIncrement(), 0);
			// 设置守护线程属性
			if (thread.isDaemon()) {
				thread.setDaemon(false);
			}
			// 设置线程优先级
			if (thread.getPriority() != Thread.NORM_PRIORITY) {
				thread.setPriority(Thread.NORM_PRIORITY);
			}
			return thread;
		}
	}
}
